package com.shujia.init

import org.apache.flink.streaming.api.scala._
import org.apache.flink.table.api.EnvironmentSettings
import org.apache.flink.table.api.bridge.scala.StreamTableEnvironment
import org.apache.flink.table.catalog.hive.HiveCatalog

/**
  * 同样的myhive的catalog，整合hive连接mysql构建ads层的三张表：性别占比表、词云热度表、微博来源表
  */
object InitFLinkAdsTable {
  def main(args: Array[String]): Unit = {


    val bsEnv: StreamExecutionEnvironment = StreamExecutionEnvironment.getExecutionEnvironment

    val bsSettings: EnvironmentSettings = EnvironmentSettings
      .newInstance()
      .useBlinkPlanner() //使用blink计划器
      .inStreamingMode() //流处理模型
      .build()


    //创建table环境
    val bsTableEnv: StreamTableEnvironment = StreamTableEnvironment.create(bsEnv, bsSettings)

    val hiveCatalog: HiveCatalog = new HiveCatalog("myHive","sent","sentcompute/src/main/resources")

    bsTableEnv.registerCatalog("myHive",hiveCatalog)

    bsTableEnv.useCatalog("myHive")


    bsTableEnv.executeSql(
      """
        |use ads
        |
      """.stripMargin)


    bsTableEnv.executeSql(
      """
        |drop table IF EXISTS  ads_mysql_gender_index
        |
      """.stripMargin)
    //连接mysql构建性别占比表
    bsTableEnv.executeSql(
      """
        |
        |CREATE TABLE ads_mysql_gender_index (
        | gender String,
        | c bigint,
        | PRIMARY KEY (gender) NOT ENFORCED
        |) WITH (
        |   'connector' = 'jdbc',
        |   'url' = 'jdbc:mysql://master:3306/sent',
        |   'table-name' = 'ads_mysql_gender_index',
        |   'username' = 'root',
        |   'password' = '123456'
        |)
        |
      """.stripMargin)




    bsTableEnv.executeSql(
      """
        |drop table IF EXISTS  ads_mysql_word_index
        |
      """.stripMargin)
    //连接mysql构建词云热度表
    bsTableEnv.executeSql(
      """
        |
        |CREATE TABLE ads_mysql_word_index (
        | word String,
        | c bigint,
        | PRIMARY KEY (word) NOT ENFORCED
        |) WITH (
        |   'connector' = 'jdbc',
        |   'url' = 'jdbc:mysql://master:3306/sent?useUnicode=true&characterEncoding=UTF-8',
        |   'table-name' = 'ads_mysql_word_index',
        |   'username' = 'root',
        |   'password' = '123456'
        |)
        |
      """.stripMargin)



    bsTableEnv.executeSql(
      """
        |drop table IF EXISTS  ads_mysql_source_index
        |
      """.stripMargin)
    //连接mysql构建微博来源表
    bsTableEnv.executeSql(
      """
        |
        |CREATE TABLE ads_mysql_source_index (
        | source String,
        | c bigint,
        | PRIMARY KEY (source) NOT ENFORCED
        |) WITH (
        |   'connector' = 'jdbc',
        |   'url' = 'jdbc:mysql://master:3306/sent?useUnicode=true&characterEncoding=UTF-8',
        |   'table-name' = 'ads_mysql_source_index',
        |   'username' = 'root',
        |   'password' = '123456'
        |)
        |
      """.stripMargin)




    bsTableEnv.executeSql(
      """
        |drop table IF EXISTS  ads_mysql_weibo_topN
        |
      """.stripMargin)

    bsTableEnv.executeSql(
      """
        |
        |CREATE TABLE ads_mysql_weibo_topN (
        | id String,
        | url String,
        | text String,
        | score bigint,
        | PRIMARY KEY (id) NOT ENFORCED
        |) WITH (
        |   'connector' = 'jdbc',
        |   'url' = 'jdbc:mysql://master:3306/sent?useUnicode=true&characterEncoding=UTF-8',
        |   'table-name' = 'ads_mysql_weibo_topN',
        |   'username' = 'root',
        |   'password' = '123456'
        |)
        |
      """.stripMargin)






  }
}
